package com.leadbank.bigdata.hbase;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.Iterator;
import java.util.List;

/**
 * Created by hp on 2018/4/24.
 */
public class HbaseManagerMain {

    private static final Log LOG = LogFactory.getLog(HbaseManagerMain.class);

    private static final String TABLE_NAME = "m_domain";
    private static final String COLUMN_FAMILY_NAME = "cf";

    private Configuration configuration;
    private Connection connection;
    private Admin admin;

    @Before
    public void before() {
        configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.zookeeper.quorum", "10.1.13.94,10.1.13.97,10.1.13.99");
        configuration.set("hbase.master", "10.1.13.94");

        try {
            connection = ConnectionFactory.createConnection(configuration);
            admin = connection.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 列出表
     */
    @Test
    public void listTables() throws IOException {
        TableName[] names = admin.listTableNames();
        for (TableName tableName : names) {
            LOG.info("Table Name is : " + tableName.getNameAsString());
            System.out.println("Table Name is : " + tableName.getNameAsString());
        }
    }

    /**
     * 判断表是否存在
     */
    @Test
    public void isTableExists() throws IOException {
        /**
         * org.apache.hadoop.hbase.TableName为为代表了表名字的Immutable POJO class对象,
         * 形式为<table namespace>:<table qualifier>。
         *   static TableName  valueOf(byte[] fullName)
         *  static TableName valueOf(byte[] namespace, byte[] qualifier)
         *  static TableName valueOf(ByteBuffer namespace, ByteBuffer qualifier)
         *  static TableName valueOf(String name)
         *  static TableName valueOf(String namespaceAsString, String qualifierAsString)
         * HBase系统默认定义了两个缺省的namespace
         *     hbase：系统内建表，包括namespace和meta表
         *     default：用户建表时未指定namespace的表都创建在此
         * 在HBase中，namespace命名空间指对一组表的逻辑分组，类似RDBMS中的database，方便对表在业务上划分。
         *
         */
        TableName tableName = TableName.valueOf(TABLE_NAME);

        boolean exists = admin.tableExists(tableName);
        if (exists) {
            LOG.info("Table " + tableName.getNameAsString() + " already exists.");
            System.out.println("Table " + tableName.getNameAsString() + " already exists.");
        } else {
            LOG.info("Table " + tableName.getNameAsString() + " not exists.");
            System.out.println("Table " + tableName.getNameAsString() + " not exists.");
        }
    }

    /**
     * 删除表
     * @throws IOException
     */
    @Test
    public void deleteTable() throws IOException {
        TableName tableName = TableName.valueOf(TABLE_NAME);
        LOG.info("disable and then delete table named " + TABLE_NAME);
        System.out.println("disable and then delete table named " + TABLE_NAME);
        admin.disableTable(tableName);
        admin.deleteTable(tableName);
    }



    /**
     * 创建表
     * @throws IOException
     */
    @Test
    public void createTable() throws IOException {
        TableName tableName = TableName.valueOf(TABLE_NAME);
        HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
        HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(COLUMN_FAMILY_NAME);
        hTableDescriptor.addFamily(hColumnDescriptor);

        admin.createTable(hTableDescriptor);
    }

    /**
     * 添加数据
     * @throws IOException
     */
    @Test
    public void putDatas() throws IOException {
        String [] rows = {"baidu.com_19991011_20151011", "alibaba.com_19990415_20220523"};
        String [] columns = {"owner", "ipstr", "access_server", "reg_date", "exp_date"};
        String [][] values = {
                {"Beijing Baidu Technology Co.", "220.181.57.217", "北京", "3999年10月11日", "4015年10月11日"},
                {"Hangzhou Alibaba Advertising Co.", "205.204.101.42", "杭州", "3999年04月15日", "4022年05月23日"}
        };
        Table table = null;
        try {
            TableName tableName = TableName.valueOf(TABLE_NAME);
            table = connection.getTable(tableName);
            byte[] family = Bytes.toBytes(COLUMN_FAMILY_NAME);
            for (int i = 0; i < rows.length; i++) {
                System.out.println("========================" + rows[i]);
                byte[] rowkey = Bytes.toBytes(rows[i]);
                Put put = new Put(rowkey);
                for (int j = 0; j < columns.length; j++) {
                    byte[] qualifier = Bytes.toBytes(columns[j]);
                    byte[] value = Bytes.toBytes(values[i][j]);
                    put.addColumn(family, qualifier, value);
                }
                table.put(put);
            }
        } finally {
            table.close();
        }

    }

    /**
     * 检索数据-单行获取
     * @throws IOException
     */
    @Test
    public void getData() throws IOException {
        LOG.info("Get data from table " + TABLE_NAME + " by family.");
        System.out.println("Get data from table " + TABLE_NAME + " by family.");

        TableName tableName = TableName.valueOf(TABLE_NAME);
        byte[] family = Bytes.toBytes(COLUMN_FAMILY_NAME);
        byte[] row = Bytes.toBytes("baidu.com_19991011_20151011");
        Table table = connection.getTable(tableName);

        Get get = new Get(row);
        get.addFamily(family);
        // 也可以通过addFamily或addColumn来限定查询的数据
        Result result = table.get(get);
        List<Cell> cells = result.listCells();
        for(Cell cell : cells) {
            String qualifier = new String(CellUtil.cloneQualifier(cell));
            String value = new String(CellUtil.cloneValue(cell));
            // @Deprecated
            // LOG.info(cell.getQualifier() + "\t" + cell.getValue());
            LOG.info(qualifier + "\t" + value);
            System.out.println(qualifier + "\t" + value);
        }
    }

    /**
     * 检索数据-表扫描
     * @throws IOException
     */
    @Test
    public void scanTable() throws IOException {
        LOG.info("Scan table " + TABLE_NAME + " to browse all datas.");
        System.out.println("Scan table " + TABLE_NAME + " to browse all datas.");
        TableName tableName = TableName.valueOf(TABLE_NAME);
        byte[] family = Bytes.toBytes(COLUMN_FAMILY_NAME);

        Scan scan = new Scan();
        scan.addFamily(family);

        Table table = connection.getTable(tableName);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Iterator<Result> it = resultScanner.iterator(); it.hasNext();) {
            Result result = it.next();
            List<Cell> cells = result.listCells();
            String row = new String(result.getRow());
            for(Cell cell : cells) {
                String qualifier = new String(CellUtil.cloneQualifier(cell));
                String value = new String(CellUtil.cloneValue(cell));
                // @Deprecated
                // LOG.info(cell.getQualifier() + "\t" + cell.getValue());
                LOG.info(qualifier + "\t" + value);
                System.out.println(row + "\t" +qualifier + "\t" + value);
            }
        }
    }

    /**
     * 安装条件检索数据
     */
    @Test
    public void queryByFilter() throws IOException {
        // 简单分页过滤器示例程序
        Filter filter = new PageFilter(1);

        TableName tableName = TableName.valueOf(TABLE_NAME);
        byte[] family = Bytes.toBytes(COLUMN_FAMILY_NAME);

        Scan scan = new Scan();
        scan.setFilter(filter);
        scan.addFamily(family);

        Table table = connection.getTable(tableName);
        ResultScanner resultScanner = table.getScanner(scan);
        for(Iterator<Result> it = resultScanner.iterator(); it.hasNext();) {
            Result result = it.next();
            List<Cell> cells = result.listCells();
            String row = new String(result.getRow());
            for(Cell cell : cells) {
                String qualifier = new String(CellUtil.cloneQualifier(cell));
                String value = new String(CellUtil.cloneValue(cell));
                // @Deprecated
                // LOG.info(cell.getQualifier() + "\t" + cell.getValue());
                LOG.info(qualifier + "\t" + value);
                System.out.println(row + "\t" +qualifier + "\t" + value);
            }
        }
    }

    /**
     * 删除数据
     * @param connection
     * @throws IOException
     */
    @Test
    public void deleteDatas() throws IOException {
        LOG.info("delete data from table " + TABLE_NAME + " .");
        System.out.println("delete data from table " + TABLE_NAME + " .");
        TableName tableName = TableName.valueOf(TABLE_NAME);
        byte[] family = Bytes.toBytes(COLUMN_FAMILY_NAME);
        byte[] row = Bytes.toBytes("baidu.com_19991011_20151011");
        Delete delete = new Delete(row);

        // @deprecated Since hbase-1.0.0. Use {@link #addColumn(byte[], byte[])}
        // delete.deleteColumn(family, qualifier);            // 删除某个列的某个版本
        delete.addColumn(family, Bytes.toBytes("owner"));

        // @deprecated Since hbase-1.0.0. Use {@link #addColumns(byte[], byte[])}
        // delete.deleteColumns(family, qualifier)            // 删除某个列的所有版本

        // @deprecated Since 1.0.0. Use {@link #(byte[])}
        // delete.addFamily(family);                           // 删除某个列族

        Table table = connection.getTable(tableName);
        table.delete(delete);
    }


    @After
    public void after() {
        try {
            if (null != admin)
                admin.close();
            if (null != connection)
                connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }
}
